
FILEBASE_ROOT = 'E:\EyePoint' # путь к рабочей папке программы


import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import itertools
from sklearn import __version__ as sk_version
from scipy.signal import resample

import PySpice
import PySpice.Logging.Logging as Logging
from PySpice.Spice.Netlist import Circuit
from PySpice.Spice.Library import SpiceLibrary
from PySpice.Doc.ExampleTools import find_libraries
from PySpice.Unit import *

import os
import uuid
import random
import requests
import zipfile
import json

plt.style.use('seaborn')
plt.rcParams["figure.figsize"] = (10,8)

logger = Logging.setup_logging()
PySpice.show_version()

print('sklearn version '+sk_version)

Circuit_Set = set() # set of all availible curcuits

PySpice Version 1.4.3
sklearn version 1.0.2


def download_and_unzip(filePath):
    print('downloading started..')
                       
    url = 'https://github.com/vasily84/PySpice_CurveTracer/raw/main/spice_library.zip'
    # Downloading the file by sending the request to the URL
    req = requests.get(url)

    zipName = os.path.join(filePath,'temp.zip')
    print('from '+str(url)+'\ndownloaded '+str(len(req.content))+' bytes')
    
    # Writing the file to the local file system
    with open(zipName,'wb') as output_file:
        output_file.write(req.content)
    print('downloading completed, UnZip started..')
    
    # unzip 
    with zipfile.ZipFile(zipName, 'r') as zip_ref:
        zip_ref.extractall(filePath)
        
    os.remove(zipName)
    print('UnZip completed..')
    
    
# проверяем пути к рабочей папке,
# при необходимости подгружаем spice библиотеки компонентов
# и обученные модели с github
if not os.path.exists(FILEBASE_ROOT):
    os.makedirs(FILEBASE_ROOT)
    download_and_unzip(FILEBASE_ROOT)

print('\nsimulation dataset on:\n'+FILEBASE_ROOT)

libraries_path = os.path.join(FILEBASE_ROOT,'spice_library')
print('\nspice models library on:\n'+libraries_path)
spice_library = SpiceLibrary(libraries_path)
print(spice_library['1N4148'])

simulation dataset on:
E:\EyePoint

spice models library on:
E:\EyePoint\spice_library
E:\EyePoint\spice_library\diode\switching\1N4148.lib


class CurveTracer_Settings():
    """ EyePoint hardware settings struct"""
    def __init__(self,F=100,V=5.,Rcs=4750.,SNR=40.0,lendata=100,cycle=3,json=None):
        self.F = F
        self.V = V
        self.Rcs = Rcs
        self.SNR = SNR
        self.lendata = lendata
        self.cycle = cycle
        if json is not None:
            self.from_json(json)
    
    def to_numpy(self):
        arr = np.array([self.F, self.V, self.Rcs, self.SNR, self.lendata, self.cycle])
        return arr
    
    def from_numpy(self,arr):
        self.F = arr[0]
        self.V = arr[1]
        self.Rcs = arr[2]
        self.SNR = arr[3]
        self.lendata = int(arr[4])
        self.cycle = int(arr[5])
        return self
        
    def __str__(self):
        s = 'F='+str(self.F)+' V='+str(self.V)+' lendata='+str(self.lendata)+' Rcs='+str(self.Rcs)+' cycle='+str(self.cycle)+' SNR='+str(self.SNR)
        return s
    
    def __eq__(self,obj):
        """ сравниваем по критерию - частота, напряжение, число точек,вн.сопротивление """
        lendata_ok = (self.lendata==obj.lendata)
        F_ok = abs(self.F-obj.F) <= 0.1 # точность сравнения 0.1 Гц
        V_ok = abs(self.V-obj.V) <= 0.1 # точность сравнения 0.1 Вольт
        Rcs_ok = abs(self.Rcs-obj.Rcs) <=0.1 # точность 0.1 Ом
        return (lendata_ok and F_ok and V_ok)
    
    def __hash__(self):
        """ поддержка hash для работы с множествами"""
        return hash((self.F,self.V,self.lendata,self.Rcs))
    
    def from_json(self,jsonRec):
        #print(jsonRec['iv_curves'][0]['measurement_settings'])
        self.F = jsonRec['iv_curves'][0]['measurement_settings']['probe_signal_frequency']
        self.V = jsonRec['iv_curves'][0]['measurement_settings']['max_voltage']
        self.Rcs = jsonRec['iv_curves'][0]['measurement_settings']['internal_resistance']
        sr = jsonRec['iv_curves'][0]['measurement_settings']['sampling_rate']
        self.lendata = sr//self.F
        return self
        

def CreateCVC(circuit,ct):
    """ просимулировать схему в PySpice. circuit - схема, ct - настройки симуляции"""
    cycle = ct.cycle
    lendata = ct.lendata
    period = 1 / ct.F
    rms_voltage = ct.V / np.sqrt(2)
    # добавляем испытательный щуп в схему
    circuit.R('cs', 'input', 'input_dummy', ct.Rcs)
    circuit.AcLine('Current', circuit.gnd, 'input_dummy', rms_voltage=rms_voltage, frequency=ct.F)
    
    simulator = circuit.simulator()
    analysis = simulator.transient(step_time=period / lendata, end_time=period * cycle)
    analysis.input_dummy = analysis.input_dummy[len(analysis.input_dummy)-lendata:len(analysis.input_dummy)]
    analysis.VCurrent = analysis.VCurrent[len(analysis.VCurrent)-lendata:len(analysis.VCurrent)]
# Расчитываем шум независмо для тока и напряжения исходя из среднеквадратичных значений и одинакового SNR
    avg_V_db = 10 * np.log10(np.mean(np.array(analysis.input_dummy, dtype=float) ** 2))
    avg_Vnoise_db = avg_V_db - ct.SNR
    Vnoise = np.random.normal(0, np.sqrt(10 ** (avg_Vnoise_db / 10)), len(analysis.input_dummy))
    analysis.input_dummy = np.array(analysis.input_dummy, dtype=float) + Vnoise
    avg_I_db = 10 * np.log10(np.mean(np.array(analysis.VCurrent, dtype=float) ** 2))
    avg_Inoise_db = avg_I_db - ct.SNR
    Inoise = np.random.normal(0, np.sqrt(10 ** (avg_Inoise_db / 10)), len(analysis.VCurrent))
    analysis.VCurrent = np.array(analysis.VCurrent, dtype=float) + Inoise
    # удаляем испытательный щуп из схемы
    circuit.Rcs.detach()
    circuit.VCurrent.detach()
    return analysis

class CAnalysis():
    def __init__(self,VCurrent=None,input_dummy=None,json=None):
        self.VCurrent = VCurrent
        self.input_dummy = input_dummy
        if json is not None:
            self.from_json(json)
    
    def from_json(self,jsonRec):
        self.VCurrent = jsonRec["iv_curves"][0]["currents"]
        self.input_dummy = jsonRec["iv_curves"][0]["voltages"]
        return self
        
def load_JSON_Records(filePath):
    """ загрузка с диска набора снятых прибором данных - самого сигнала и параметров его съемки """
    with open(filePath, "r") as jfile:
        J = json.load(jfile)
    return J["elements"][0]["pins"]

def plot_analysis_IV(analysis,label='',ax=None):
    """ plot I(V) curve for analysis """
    if ax is None:
        plt.plot(analysis.input_dummy, analysis.VCurrent,label=label)
        plt.title('I(V)')
        plt.xlabel('Напряжение [В]')
        plt.ylabel('Сила тока [А]')
        return
    ax.plot(analysis.input_dummy, analysis.VCurrent,label=label)
    ax.set_title('I(V)')
    ax.set_xlabel('Напряжение [В]')
    ax.set_ylabel('Сила тока [А]')


# варианты настройки пробора по умолчанию - перечисляем врукопашную 
ct_List = list()
ct = CurveTracer_Settings(F=1.,V=3.3,lendata=100,Rcs=475.)
ct_List.append(ct)
ct = CurveTracer_Settings(F=100.,V=5.,lendata=100,Rcs=4750.)
ct_List.append(ct)
ct = CurveTracer_Settings(F=100_000,V=12.,lendata=20,Rcs=47500.)
ct_List.append(ct)

try: # просканируем папку, добавим оттуда варианты настройки прибора
    path_json = os.path.join(FILEBASE_ROOT,'RealSignalFolder')
    fileList = [f for f in os.listdir(path_json) if f.endswith('.json')]
    
    # просмотрим доступные файлы, составим множество из конфигураций прибора
    for f1 in fileList:
        ff = os.path.join(FILEBASE_ROOT,'RealSignalFolder',f1)
        jRecs = load_JSON_Records(ff)
        for j in jRecs:
            ct = CurveTracer_Settings(json=j)
            #analysis = CAnalysis(json=j)
            ct_List.append(ct)
finally:
    # удалим дубликаты
    ct_List_uniq = []
    for ct in ct_List:
        if ct not in ct_List_uniq:
            ct_List_uniq.append(ct)
    ct_List = ct_List_uniq
    print('**list of curve tracer settings:')
    for ct in ct_List:
         print(ct)

**list of curve tracer settings:
F=1.0 V=3.3 lendata=100 Rcs=475.0 cycle=3 SNR=40.0
F=100.0 V=5.0 lendata=100 Rcs=4750.0 cycle=3 SNR=40.0
F=100000 V=12.0 lendata=20 Rcs=47500.0 cycle=3 SNR=40.0


%%script false
%time
circuit = Circuit('RD')
circuit.include(spice_library['1N4148'])
circuit.R(1,'input',1,10@u_kOhm)
circuit.X('D1', '1N4148', 1, circuit.gnd)
circuit.R(2,'input',circuit.gnd,20@u_kOhm)

analysis = CreateCVC(circuit,CurveTracer_Settings())
        
#plt.grid()
plt.plot(analysis.input_dummy, analysis.VCurrent)
plt.xlabel('Напряжение [В]')
plt.ylabel('Сила тока [А]')
plt.show()

Couldn't find program: 'false'


def save_cir_analysis(fileName,circuit_params,curve_params,analysis):
    """ save circuit nominals,curve tracer parameters, voltage and current arrays to file"""
    with open(fileName, 'wb') as f:
        np.save(f,circuit_params)
        np.save(f,curve_params)
        np.save(f,analysis.VCurrent)
        np.save(f,analysis.input_dummy)
    
def make_sure_path(path_title):
    """ create path if path not exist, return full path """
    dirName = os.path.join(FILEBASE_ROOT,path_title)
    if not os.path.exists(dirName):
        os.makedirs(dirName)
    return dirName

def save_circuit(dirName,circuit):
    """ save sample circuit on disk """
    fileName = os.path.join(dirName,circuit.title+'.cir')
    with open(fileName,'w') as f:
        f.write(str(circuit))


def load_circuit_dataset(path_dataset,ct=None,shuffle=True,max_files=None):
    """ load dataset from circuit folder.
    return X,y """
    fileList = [f for f in os.listdir(path_dataset) if f.endswith('.npy')]
    
    if shuffle: 
        random.shuffle(fileList)
        
    if not max_files is None:
        max_files = min(max_files,len(fileList))
        fileList=fileList[0:max_files]
    
    X_data,y_data=[],[]
    
    for fileName in tqdm(fileList): # process files
        with open(os.path.join(path_dataset,fileName), 'rb') as f:
            circuit_params = np.load(f)
            curve_params = np.load(f)
            VCurrent = np.load(f)
            input_dummy = np.load(f)  
                                
            if not ct is None: 
                ct0 = CurveTracer_Settings().from_numpy(curve_params)
                if not (ct0==ct):
                    continue # пропускаем сигналы с неподходящими параметрами
                    
        # append data to arrays
        y_data.append(circuit_params)
        X_data.append([VCurrent,input_dummy])
        
    return np.array(X_data),np.array(y_data)


from sklearn.preprocessing import FunctionTransformer,StandardScaler
from scipy.fft import rfft

def my_xgb_transform_function(X):
    """ feature engineering on raw analysis signal data.
    Transformations - fft image, input signal resampling etc.""" 
    Xa = []
    for x in X:
        I_signal = x[:len(x)//2] # current      
        V_signal = x[len(x)//2:] # input voltage - not use
        I_fft = rfft(I_signal)
        I_fft = I_fft[:20] # use only first 20 points on freq domain
        I_abs = np.abs(I_fft)
        I_angle = np.angle(I_fft)
        I_real = np.real(I_fft)
        I_imag = np.imag(I_fft)
        
        V_fft = rfft(V_signal)
        V_fft = V_fft[:20] # use only first 20 points on freq domain
        V_abs = np.abs(V_fft)
        V_angle = np.angle(V_fft)
        V_real = np.real(V_fft)
        V_imag = np.imag(V_fft)

        x1 = np.hstack((x,I_abs,I_angle,I_real,I_imag,V_abs,V_angle,V_real,V_imag))
        Xa.append(x1)
        
    return np.array(Xa)

def my_rfft_transform_function(X):
    """ feature engineering on raw analysis signal data.
    Transformations - fft image, input signal resampling etc.""" 
    Xa = []
    for x in X:
        I_signal = x[:len(x)//2] # current      
        V_signal = x[len(x)//2:] # input voltage - not use
        I_fft = rfft(I_signal)
        
        I_abs = np.abs(I_fft)
        I_angle = np.angle(I_fft)
        I_real = np.real(I_fft)
        I_imag = np.imag(I_fft)
        
        V_fft = rfft(V_signal)
        
        V_abs = np.abs(V_fft)
        V_angle = np.angle(V_fft)
        V_real = np.real(V_fft)
        V_imag = np.imag(V_fft)

        x1 = np.hstack((I_abs,I_angle,I_real,I_imag,V_abs,V_angle,V_real,V_imag))
        Xa.append(x1)
        
    return np.array(Xa)

def my_linear_transform_function(X):
    """ feature engineering on raw analysis signal data.
    Transformations - fft image, input signal resampling etc.""" 
    Xa = []
    for x in X:
        I_signal = x[:len(x)//2] # current      
        V_signal = x[len(x)//2:] # input voltage - not use
        Q_signal = np.cumsum(I_signal)
        dI_signal = np.diff(I_signal)
        dV_signal = np.diff(V_signal)
        
        x1 = np.hstack((I_signal,V_signal,Q_signal,dI_signal,dV_signal))
        Xa.append(x1)
        
    return np.array(Xa)

MyXGBTransformer = FunctionTransformer(my_xgb_transform_function)
MyRfftTransformer = FunctionTransformer(my_rfft_transform_function)
MyLinearTransformer = FunctionTransformer(my_linear_transform_function)


import pickle
from scipy.optimize import minimize
from sklearn.metrics import mean_squared_error
from sklearn.linear_model import LinearRegression, LassoCV,RidgeCV
from sklearn.svm import SVR
from sklearn.ensemble import GradientBoostingRegressor, ExtraTreesRegressor,RandomForestRegressor
from sklearn.multioutput import RegressorChain
from sklearn.pipeline import make_pipeline
from copy import deepcopy

class BaseCircuitModel():
    """ base class for circuit models. Provide training, data preprocessing,
    prediction for nominal values """
    def __init__(self,title='baseAbstract',modelDir=None,modelName=None,estimator=None,ct=None):
        self.title = title
        self.estimator = estimator
        self.modelDir = os.path.join(FILEBASE_ROOT,self.title) if (modelDir is None) else modelDir  
        make_sure_path(self.modelDir)
        self.modelName = title+'.pickle' if (modelName is None) else modelName   
        self.modelFileName = os.path.join(self.modelDir,self.modelName)
        self.optimize_save_flag = False
        
        if ct is None:
            ct = CurveTracer_Settings()
            
        # deepcopy необходим, т.к. при загрузке с диска self.ct пересоздается 
        self.ct = deepcopy(ct) 
        
        try:
            self.load_model()
            return 
        except OSError as e:
            print(e)
            
        print('rebuild model ...')
        self.build_model()
        
    
    def load_model(self):
        with open(self.modelFileName, "rb") as f:
                self.model = pickle.load(f)
                self.rank = self.model.rank # number of circuit variative nominals
                self.ct = CurveTracer_Settings().from_numpy(self.model.ct)
        
    def save_model(self):
        with open(self.modelFileName, "wb") as f:
                self.model.rank = self.rank
                self.model.ct = self.ct.to_numpy()
                pickle.dump(self.model, f)
        
    def write_dataset(self,Xvals):
        """ circuit batch simulation over params Xvals"""
        for xval in tqdm(Xvals): # walk over nominals in circuit and create I/V curves
            cir_params = np.array(xval)
            circuit = self.build_circuit(cir_params)
            analysis = CreateCVC(circuit,ct=self.ct)
            fileName = os.path.join(self.modelDir,uuid.uuid4().hex+'.npy')
            save_cir_analysis(fileName,cir_params,self.ct.to_numpy(),analysis)

    def load_dataset(self):
        """ load dataset from disk"""
        X_data,y_data = load_circuit_dataset(self.modelDir,ct=self.ct)
        ADim = X_data.shape[0]
        X_data = X_data.reshape(ADim,-1)
        if self.rank==1: # fit shape 1d regression or Nd regression 
            y_data = y_data.reshape(-1)
        else:
            y_data = y_data.reshape(ADim,-1)
            
        return X_data,y_data
        
    def build_model(self,save_model=True, test_model=True ): 
        print('dataset for model '+self.title+' is loading, please wait..')
        
        if self.estimator is None:
            baseEstimator = GradientBoostingRegressor()
            
            if self.rank>1:
                self.model = make_pipeline(MyXGBTransformer,RegressorChain(baseEstimator))
            else:
                self.model = make_pipeline(MyXGBTransformer,baseEstimator)
        else:
            self.model = self.estimator
        
        try:
            X_data,y_data = self.load_dataset()
        except BaseException as e:
            print(e)
            print('rebuild dataset ...')
            self.rebuild_dataset()
            X_data,y_data = self.load_dataset()
        
        print('model '+self.title+' is training, please wait..')
        self.model.fit(X_data,y_data)
        self.model.rank = self.rank
        
        if save_model:
            self.save_model()
            self.load_model()
        
        if test_model:
            print('test_model() routine for '+self.title+' model:')
            self.test_model()
    
    def test_model(self):
        print('!!WARNING : test_model() routine for '+self.title+' model not implemented')
            
    def predict_from_model(self,analysis0,start_minimization=False):
        """ predict X values for circuit, best fitted to analysis0 """
        if not len(analysis0.VCurrent)==self.ct.lendata:
            print('len(A)='+str(len(analysis0.VCurrent)),end='')
            print(' len(B)='+str(self.ct.lendata))
            print('do resampling')
            VCurrent = resample(analysis0.VCurrent,self.ct.lendata)
            input_dummy = resample(analysis0.input_dummy,self.ct.lendata)
            X = np.array([VCurrent,input_dummy]).reshape(-1)
        else:
            X = np.array([analysis0.VCurrent,analysis0.input_dummy]).reshape(-1)
        
        a = self.model.predict([X])[0]
        if start_minimization:
            a,_ = self.optimize_from_X(analysis0,a)
        
        return a
        
    def misfit(self,analysis0, Xvals=None,start_minimization=False):
        """ calculate mse for current from circuit from Xval simulation and from analysis"""
        if Xvals is None:
            Xvals = self.predict_from_model(analysis0)
        
        if start_minimization:
            Xvals,_ = self.optimize_from_X(analysis0,Xvals)
            
        circuit = self.build_circuit(Xvals)
        analysis1 = CreateCVC(circuit,self.ct)
        
        len_0 = len(analysis0.VCurrent)
        len_1 = len(analysis1.VCurrent)
        if not len_0==len_1:
            print('do resample..')
            len_max = max(len_o,len_1)
            A_0 = resample(analysis0.VCurrent,len_max)
            A_1 = resample(analysis0.VCurrent,len_max)
            return mean_squared_error(A_1,A_0)
            
        return mean_squared_error(analysis1.VCurrent,analysis0.VCurrent)
    
    def _optimize_func(self,Xvals):      
        circuit = self.build_circuit(Xvals)
        analysis1 = CreateCVC(circuit,self.ct)
        
        if self.optimize_save_flag:
            fileName = os.path.join(self.modelDir,'o_'+uuid.uuid4().hex+'.npy')
            save_cir_analysis(fileName,Xvals,ct_default_settings.to_numpy(),analysis1)

        return mean_squared_error(analysis1.VCurrent,self._analysis0.VCurrent)

    def optimize_from_X(self,analysis0,Xvalues = None):
        if Xvalues is None:
            Xvalues = self.predict_from_model(analysis0).reshape(-1)
        self._analysis0 = analysis0 #
        
        optimize_result = minimize(self._optimize_func,Xvalues)
        
        return (optimize_result.x, optimize_result)


%%time
######################################################################

class R_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 1
        super().__init__('R',modelName=modelName,estimator=estimator,ct=ct)
            
    def build_circuit(self,Xvals): 
        if isinstance(Xvals, float):
            r1 = Xvals
        elif isinstance(Xvals, list):
            r1 = Xvals[0]
        else: # it is numpy array
            r1 = Xvals.item()
        r1 = np.abs(r1)
        circuit = Circuit('R')
        circuit.R(1,'input',circuit.gnd,r1@u_kOhm)
        return circuit
    
    def rebuild_dataset(self):
        self.write_dataset(np.random.uniform(0.1,100,size=50))
        self.write_dataset(np.random.uniform(0.0,5,size=50))
        
        
    def test_model(self):
        print('actual r \t predicted r \t acc dr/r ')
        for r in np.array([0.1, 1, 1.1, 3., 10., 15., 22., 30., 50., 70., 75., 98., 100.]):
            analysis0 = CreateCVC(self.build_circuit(r),self.ct)
            x = self.predict_from_model(analysis0)
            acc = np.abs((x-r)/r)
            print(str(r),str(x),str(acc),sep='\t')
            
Circuit_Set.add(R_circuit('R1',ct=ct_List[0]))
Circuit_Set.add(R_circuit('R2',ct=ct_List[1]))
Circuit_Set.add(R_circuit('R3',ct=ct_List[2]))


######################################################################

class C_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 1
        super().__init__('C',modelName=modelName,estimator=estimator,ct=ct)
        
    def build_circuit(self,Xvals): 
        if isinstance(Xvals, float):
            c1 = Xvals
        elif isinstance(Xvals, list):
            c1 = Xvals[0]
        else: # it is numpy array
            c1 = Xvals.item()
        c1 = np.abs(c1)
        circuit = Circuit('C')
        circuit.C(1,'input',circuit.gnd,c1@u_uF)
        return circuit
    
    def rebuild_dataset(self):
        self.write_dataset(np.random.uniform(1e-3,100.,size=50))
        
    def test_model(self):
        print('actual c \t predicted c \t acc dc/c ')
        
        for c in np.array([0.1, 1, 1.1, 3., 10., 15., 22., 30., 50., 70., 75., 98., 100.]):
            analysis0 = CreateCVC(self.build_circuit(c),self.ct)
            x = self.predict_from_model(analysis0)
            acc = np.abs((x-c)/c)
            print(str(c),str(x),str(acc),sep='\t')
    
Circuit_Set.add(C_circuit('C1',ct=ct_List[0]))
Circuit_Set.add(C_circuit('C2',ct=ct_List[1]))
Circuit_Set.add(C_circuit('C3',ct=ct_List[2]))

    
######################################################################

class L_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 1
        super().__init__('L',modelName=modelName,estimator=estimator,ct=ct)
            
    def build_circuit(self,Xvals): 
        if isinstance(Xvals, float):
            l1 = Xvals
        elif isinstance(Xvals, list):
            l1 = Xvals[0]
        else: # it is numpy array
            l1 = Xvals.item()
        l1 = np.abs(l1)
        circuit = Circuit('L')
        circuit.L(1,'input',circuit.gnd,l1@u_uH)
        return circuit
    
    def rebuild_dataset(self):
        self.write_dataset(np.random.uniform(1e-3,500,size=1000))
        
    def test_model(self):
        print('actual c \t predicted c \t acc dc/c ')
        
        for l in np.array([0.1, 1, 1.1, 3., 10., 15., 22., 30., 50., 70., 175., 398., 700.]):
            analysis0 = CreateCVC(self.build_circuit(l),self.ct)
            x = self.predict_from_model(analysis0)
            acc = np.abs((x-l)/l)
            print(str(l),str(x),str(acc),sep='\t')
        
Circuit_Set.add(L_circuit('L1',ct=ct_List[0]))
Circuit_Set.add(L_circuit('L2',ct=ct_List[1]))
Circuit_Set.add(L_circuit('L3',ct=ct_List[2]))

######################################################################

class DR_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 1
        super().__init__('DR',modelName=modelName,estimator=estimator,ct=ct)
            
    def build_circuit(self,Xvals): 
        if isinstance(Xvals, float):
            r1 = Xvals
        elif isinstance(Xvals, list):
            r1 = Xvals[0]
        else: # it is numpy array
            r1 = Xvals.item()
        r1 = np.abs(r1)
        circuit = Circuit('DR')
        circuit.include(spice_library['1N4148'])
        circuit.X('D1', '1N4148', 'input', circuit.gnd)
        circuit.R(1,'input',circuit.gnd,r1@u_kOhm)
        return circuit
    
    def rebuild_dataset(self):
        self.write_dataset(np.random.uniform(0.1,100,size=50))
        self.write_dataset(np.random.uniform(0.0,5,size=50))
        
    def test_model(self):
        print('actual r \t predicted r \t acc dr/r ')
        
        for r in np.array([0.1, 1, 1.1, 3., 10., 15., 22., 30., 50., 70., 75., 98., 100.]):
            analysis0 = CreateCVC(self.build_circuit(r),self.ct)
            x = self.predict_from_model(analysis0)
            acc = np.abs((x-r)/r)
            print(str(r),str(x),str(acc),sep='\t')
    
Circuit_Set.add(DR_circuit('DR1',ct=ct_List[0]))
Circuit_Set.add(DR_circuit('DR2',ct=ct_List[1]))
Circuit_Set.add(DR_circuit('DR3',ct=ct_List[2]))

######################################################################

class D_R_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 1
        super().__init__('D_R',modelName=modelName,estimator=estimator,ct=ct)  
        
    def build_circuit(self,Xvals): 
        if isinstance(Xvals, float):
            r1 = Xvals
        elif isinstance(Xvals, list):
            r1 = Xvals[0]
        else: # it is numpy array
            r1 = Xvals.item()
        r1 = np.abs(r1)
        circuit = Circuit('D_R')
        circuit.include(spice_library['1N4148'])
        circuit.X('D1', '1N4148', 'input', 2)
        circuit.R(1,2,circuit.gnd,r1@u_kOhm)
        return circuit
    
    def rebuild_dataset(self):
        self.write_dataset(np.random.uniform(0.1,100,size=50))
        self.write_dataset(np.random.uniform(0.0,5,size=50))
        
    def test_model(self):
        print('actual r \t predicted r \t acc dr/r ')
        
        for r in np.array([0.1, 1, 1.1, 3., 10., 15., 22., 30., 50., 70., 75., 98., 100.]):
            analysis0 = CreateCVC(self.build_circuit(r),self.ct)
            x = self.predict_from_model(analysis0)
            acc = np.abs((x-r)/r)
            print(str(r),str(x),str(acc),sep='\t')
    
Circuit_Set.add(D_R_circuit('D_R1',ct=ct_List[0]))
Circuit_Set.add(D_R_circuit('D_R2',ct=ct_List[1]))
Circuit_Set.add(D_R_circuit('D_R3',ct=ct_List[2]))

###################################################################

class RC_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 2
        super().__init__('RC',modelName=modelName,estimator=estimator,ct=ct)
        
    def build_circuit(self,Xvals):
        Xvals = np.abs(Xvals)
        r1,c1 = Xvals[0],Xvals[1]
        
        circuit = Circuit('RC')
        circuit.R(1,'input',circuit.gnd,r1@u_kOhm)
        circuit.C(1,'input',circuit.gnd,c1@u_uF)
        return circuit
        
    def rebuild_dataset(self):
        X = np.random.uniform(low=[0.1,1e-3],high=[100.,100.],size=(1000,2))
        self.write_dataset(X)
        
    def test_model(self):
        print('real r','real c','r?','c?','real rc','rc?',sep='\t')

        R = [0.1, 2. , 5. , 10. , 70.]
        C = [1., 0.1, 10., 30., 5.]
        for r,c in zip(R,C):
            analysis0 = CreateCVC(self.build_circuit([r,c]),self.ct)
            rx,cx = self.predict_from_model(analysis0)
            rc = r*c
            rcx = rx*cx
            print(str(r),str(c),str(rx),str(cx),str(rc),str(rcx),sep='\t')
    
Circuit_Set.add(RC_circuit('RC1',ct=ct_List[0]))
Circuit_Set.add(RC_circuit('RC2',ct=ct_List[1]))
Circuit_Set.add(RC_circuit('RC3',ct=ct_List[2]))

###################################################################

class R_C_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 2
        super().__init__('R_C',modelName=modelName,estimator=estimator,ct=ct)
            
    def build_circuit(self,Xvals):
        Xvals = np.abs(Xvals)
        r1,c1 = Xvals[0],Xvals[1]
        
        circuit = Circuit('R_C')
        circuit.R(1,'input',2,r1@u_kOhm)
        circuit.C(1,2, circuit.gnd,c1@u_uF)
        return circuit
    
    def rebuild_dataset(self):
        X = np.random.uniform(low=[0.1,1e-3],high=[100.,100.],size=(1000,2))
        self.write_dataset(X)
        
    def test_model(self):
        print('real r','real c','r?','c?','real rc','rc?',sep='\t')

        R = [0.1, 2. , 5. , 10. , 70.]
        C = [1., 0.1, 10., 30., 5.]
        for r,c in zip(R,C):
            analysis0 = CreateCVC(self.build_circuit([r,c]),self.ct)
            rx,cx = self.predict_from_model(analysis0)
            rc = r*c
            rcx = rx*cx
            print(str(r),str(c),str(rx),str(cx),str(rc),str(rcx),sep='\t')
    
Circuit_Set.add(R_C_circuit('R_C_1',ct=ct_List[0]))
Circuit_Set.add(R_C_circuit('R_C_2',ct=ct_List[1]))
Circuit_Set.add(R_C_circuit('R_C_3',ct=ct_List[2]))

###################################################################

class DR_R_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 2
        super().__init__('DR_R',modelName=modelName,estimator=estimator,ct=ct)
        
    def build_circuit(self,Xvals): 
        Xvals = np.abs(Xvals)
        r1,r2 = Xvals[0],Xvals[1]
        
        circuit = Circuit('DR_R')
        circuit.include(spice_library['1N4148'])
        circuit.X('D1', '1N4148', 'input', 2)
        circuit.R(1,'input',2,r1@u_kOhm)
        circuit.R(2,2,circuit.gnd,r2@u_kOhm)
        
        return circuit

    
    def rebuild_dataset(self):
        X = np.random.uniform(low=[0.1,0.1],high=[100.,100.],size=(1000,2))
        self.write_dataset(X)
        
    def test_model(self):
        print('real r1','real r2','r1?','r2?',sep='\t')

        R1 = [0.1, 2. , 5. , 10. , 70.]
        R2 = [1., 0.1, 10., 30., 5.]
        for r1,r2 in zip(R1,R2):
            analysis0 = CreateCVC(self.build_circuit([r1,r2]),self.ct)
            r1x,r2x = self.predict_from_model(analysis0)
            print(str(r1),str(r2),str(r1x),str(r2x),sep='\t')
    
Circuit_Set.add(DR_R_circuit('DR_R1',ct=ct_List[0]))
Circuit_Set.add(DR_R_circuit('DR_R2',ct=ct_List[1]))
Circuit_Set.add(DR_R_circuit('DR_R3',ct=ct_List[2]))

    
###################################################################

class DR_DR_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 2
        super().__init__('DR_DR',modelName=modelName,estimator=estimator,ct=ct)
        
    def build_circuit(self,Xvals): 
        Xvals = np.abs(Xvals)
        r1,r2 = Xvals[0],Xvals[1]
        
        circuit = Circuit('DR_DR')
        circuit.include(spice_library['1N4148'])
        circuit.X('D1', '1N4148', 'input', 2)
        circuit.R(1,2,circuit.gnd,r1@u_kOhm)
        circuit.X('D2', '1N4148', 3, 'input')
        circuit.R(2,3,circuit.gnd,r2@u_kOhm)
        
        return circuit

    
    def rebuild_dataset(self):
        X = np.random.uniform(low=[0.1,0.1],high=[100.,100.],size=(1000,2))
        self.write_dataset(X)
        
        
    def test_model(self):
        print('r1','r2','r1?','r2?',sep='\t')

        R1 = [0.1, 2. , 5. , 10. , 70.]
        R2 = [1., 0.1, 10., 30., 5.]
        for r1,r2 in zip(R1,R2):
            analysis0 = CreateCVC(self.build_circuit([r1,r2]),self.ct)
            r1x,r2x = self.predict_from_model(analysis0)
            print(str(r1),str(r2),str(r1x),str(r2x),sep='\t')
    
Circuit_Set.add(DR_DR_circuit('DR_DR1',ct=ct_List[0]))
Circuit_Set.add(DR_DR_circuit('DR_DR2',ct=ct_List[1]))
Circuit_Set.add(DR_DR_circuit('DR_DR3',ct=ct_List[2]))

Wall time: 552 ms


%%script false
%%time
# тестирование разных вариантов моделей
myRegr = make_pipeline(MyLinearTransformer,GradientBoostingRegressor())
Circuit_Set.add(R_circuit('R_1.pickle',myRegr))

myRegr = make_pipeline(MyRfftTransformer,GradientBoostingRegressor())
Circuit_Set.add(R_circuit('R_2.pickle',myRegr))

myRegr = make_pipeline(MyLinearTransformer,RegressorChain(GradientBoostingRegressor()))
Circuit_Set.add(RC_circuit('RC_1.pickle',myRegr))

myRegr = make_pipeline(MyRfftTransformer,RegressorChain(GradientBoostingRegressor()))
Circuit_Set.add(RC_circuit('RC_2.pickle',myRegr))

Couldn't find program: 'false'


%%time
###################################################################
class DR_DR_R_circuit(BaseCircuitModel):
    def __init__(self,modelName=None,estimator=None,ct=None):
        self.rank = 3
        super().__init__('DR_DR_R',modelName=modelName,estimator=estimator,ct=ct)
        
    def build_circuit(self,Xvals): 
        Xvals = np.abs(Xvals)
        r1,r2,r3 = Xvals[0],Xvals[1],Xvals[2]
        
        circuit = Circuit('DR_DR_R')
        circuit.include(spice_library['1N4148'])
        circuit.X('D1', '1N4148', 'input', 2)
        circuit.R(1,2,circuit.gnd,r1@u_kOhm)
        circuit.X('D2', '1N4148', 3, 'input')
        circuit.R(2,3,circuit.gnd,r2@u_kOhm)
        circuit.R(3,'input',circuit.gnd,r3@u_kOhm)
        
        return circuit

    
    def rebuild_dataset(self):
        X = np.random.uniform(low=[0.1,0.1,0.1],high=[100.,100.,100.],size=(1000,3))
        self.write_dataset(X)
        
        
    def test_model(self):
        print('r1','r2','r3','r1?','r2?','r3?',sep='\t')

        R1 = [0.1, 2. , 5. , 10. , 70.]
        R2 = [1., 0.1, 10., 30., 5.]
        R3 = [1., 0.1, 10., 30., 5.]
        for r1,r2,r3 in zip(R1,R2,R3):
            analysis0 = CreateCVC(self.build_circuit([r1,r2,r3]),self.ct)
            r1x,r2x,r3x = self.predict_from_model(analysis0)
            print(str(r1),str(r2),str(r3),str(r1x),str(r2x),str(r3x),sep='\t')
    
    
Circuit_Set.add(DR_DR_R_circuit('DR_DR_R1',ct=ct_List[0]))
Circuit_Set.add(DR_DR_R_circuit('DR_DR_R2',ct=ct_List[1]))
Circuit_Set.add(DR_DR_R_circuit('DR_DR_R3',ct=ct_List[2]))

Wall time: 54 ms


#EP = C_circuit()
#EP.test_model()


class CircuitPredictor():
    """ class for circuit prediction from i,v curves using list of various circuit models"""
    def __init__(self):
        # append all availible circuit models to this list 
        self.circuitModels = []
        for cir in Circuit_Set:
            self.circuitModels.append(cir)
    
    
    def predict_from_model(self,analysis0,ct=None):
        """"""
        m_List = []
        for m in self.circuitModels:
            if not ct is None:
                if not m.ct==ct:
                    continue # модели с неподходящими настройками прибора игнорируем
                    
            Xvars = m.predict_from_model(analysis0)
            err = m.misfit(analysis0,Xvars)
            m_List.append((Xvars,m,err))
            
        if len(m_List)==0:
            raise RuntimeError('model settings '+str(ct)+' not representeted in models library')
            
        m_List = sorted(m_List, key = lambda k:k[2])
        best = m_List[0] # best suited model
        return best
    
    def test_visual(self,analysis0,ct0):
        """ вывести график и наиболее близко подобранный из всех возможных"""
        Xvars, objX,_ = self.predict_from_model(analysis0,ct0)
        # строим analysis по максимально подобранной схеме
        analysis1 = CreateCVC(objX.build_circuit(Xvars),objX.ct)
        plot_analysis_IV(analysis0,'исходный')
        plot_analysis_IV(analysis1,'предсказанный')
        plt.legend()
        plt.show()
        
    def test_all_json(self):
        # просканируем папку, добавим оттуда варианты настройки прибора
        path_json = os.path.join(FILEBASE_ROOT,'RealSignalFolder')
        fileList = [f for f in os.listdir(path_json) if f.endswith('.json')]
    
        # просмотрим доступные файлы, составим множество из конфигураций прибора
        for f1 in fileList:
            ff = os.path.join(FILEBASE_ROOT,'RealSignalFolder',f1)
            jRecs = load_JSON_Records(ff)
            for j in jRecs:
                ct0 = CurveTracer_Settings(json=j)
                analysis0 = CAnalysis(json=j)
                self.test_visual(analysis0,ct0)  

    def test_me(self):
        self.test_rankN(1)
        self.test_rankN(2)
        self.test_rankN(3)
        
    def test_rankN(self,N,runsCount = 50,plot_Ok=10,plot_Fail=20):
        print('RANK'+str(N)+' CIRCUIT TESTING ..')
        CIR_N = []
        for cir in Circuit_Set:
            if cir.rank==N:
                CIR_N.append(cir)
        
        if len(CIR_N)==0:
            return 
        
        count_ok = 0
        count_fail = 0
        
        for _ in range(runsCount): 
            objCir = random.choice(CIR_N)
            x = 100*np.random.random(N)
            analysis0 = CreateCVC(objCir.build_circuit(x),objCir.ct)
            Xvars, objX,_ = self.predict_from_model(analysis0,objCir.ct)
            
            if isinstance(objX,type(objCir)):
                print('success: '+str(x)+' -> '+str(Xvars)+' circuit '+objX.title)
                count_ok += 1
                if plot_Ok>0:
                    plot_Ok -=1
                    analysis1 = CreateCVC(objX.build_circuit(Xvars),objX.ct)
                    #Xvars_2 = objX.predict_from_model(analysis0,start_minimization=True)
                    #analysis2 = CreateCVC(objX.build_circuit(Xvars_2))
                    print('SUCCESS:')
                    plot_analysis_IV(analysis0,'исходный')
                    plot_analysis_IV(analysis1,'предсказанный')
                    #plot_analysis_IV(analysis2)
                    plt.legend()
                    plt.show()
            else:
                print('!!fail:  '+str(x)+' -> '+str(Xvars)+' '+objCir.title+' detected like '+objX.title)
                count_fail += 1
                if plot_Fail>0:
                    plot_Fail -=1
                    analysis1 = CreateCVC(objX.build_circuit(Xvars),objX.ct)
                    #Xvars_2 = objX.predict_from_model(analysis0,start_minimization=True)
                    #analysis2 = CreateCVC(objX.build_circuit(Xvars_2))
                    print('FAIL:')
                    plot_analysis_IV(analysis0,'исходный')
                    plot_analysis_IV(analysis1,'предсказанный')
                    #plot_analysis_IV(analysis2)
                    plt.legend()
                    plt.show()
                
        print('\nSUCCESS: '+str(count_ok)+', FAIL: '+str(count_fail)+' of '+str(count_ok+count_fail)+' tests')


cre = CircuitPredictor()


%%time
cre.test_me()

RANK1 CIRCUIT TESTING ..
2022-02-23 12:59:14,930 - PySpice.Spice.NgSpice.Shared.NgSpiceShared - Shared.WARNING - spinit was not found
2022-02-23 12:59:14,932 - PySpice.Spice.NgSpice.Shared.NgSpiceShared - Shared.ERROR - Note: can't find init file.
success: [28.19377763] -> 27.59239218891685 circuit R
SUCCESS:

success: [53.10144701] -> 57.832848500116434 circuit C
SUCCESS:

!!fail:  [46.58821552] -> [54.40543775 49.27734698] C detected like RC
FAIL:

success: [43.07764252] -> 42.39495433304964 circuit D_R
SUCCESS:

success: [74.17847766] -> 76.58019817488747 circuit DR
SUCCESS:

success: [15.23360916] -> 18.645403066997183 circuit D_R
SUCCESS:

success: [42.196708] -> 43.76748461812928 circuit DR
SUCCESS:

!!fail:  [7.3404058] -> [1.39063693 4.26296717] DR detected like DR_DR
FAIL:

success: [97.94089366] -> 292.3708031490152 circuit L
SUCCESS:

success: [13.07307403] -> 12.637160907969458 circuit DR
SUCCESS:


%%time
cre.test_all_json()

Wall time: 16.8 s

PySpice_EyePoint_CurveTracer¶

часть проекта по восстановлению электронной схемы характериографом¶

https://github.com/vasily84/PySpice_CurveTracer ¶

Необходимы - python3.x, sklearn,PySpice¶

необходимые библиотеки -¶

PySpice - симуляция электронных схем из Python¶

scikit-learn библиотека машинного обучения¶

Как использовать?¶

Необходимо задать рабочую папку программы в переменной FILEBASE_ROOT ( ячейка ниже).¶

Варианты запуска программы :¶

Код написан последовательно сверху вниз-¶

проверяем пути к рабочей папке, при необходимости подгружаем spice библиотеки компонентов и обученные модели с github¶

код CreateCVC базируется на библиотеке MySpice¶

пример использования CreateCVC¶

как построена база?¶

препроцессор данных для моделей -¶

делает преобразование фурье, прореживание и прочие манипуляции со входными сигналами¶

Мотивация к добавлению в признаковое пространство конкретных алгебраических величин (закон Ома):¶

Итого - параметры R, C, L линейно связаны с сигналами V, I, их производными (изменениями от точке к точке) и интегралами (суммами)¶

Мотивация к использованию логарифмов:¶

Итого - можно взять логарифмы от исходных сигналов, их производных и интегралов, и линейная модель их свяжет с логарифмами искомых параметров R, C, L.¶

базовый класс схем¶

Реализация конкретных схем из резисторов, емкостей, индуктивностей и диодов¶

rank 1, rank 2¶

Реализация конкретных схем из резисторов, емкостей, индуктивностей и диодов¶

rank 3¶

тестирование разных схем¶

Класс для предсказания схемы по ВАХ¶

тестирование класса предсказания схем¶

Синтетические тесты - искуственно сгенерированные данные¶

Тесты на реальных приборных Данных¶

PySpice_EyePoint_CurveTracer¶

часть проекта по восстановлению электронной схемы характериографом¶

https://github.com/vasily84/PySpice_CurveTracer¶

Необходимы - python3.x, sklearn,PySpice¶

необходимые библиотеки -¶

PySpice - симуляция электронных схем из Python¶

scikit-learn библиотека машинного обучения¶

Как использовать?¶

Необходимо задать рабочую папку программы в переменной FILEBASE_ROOT ( ячейка ниже).¶

Варианты запуска программы :¶

Код написан последовательно сверху вниз-¶

проверяем пути к рабочей папке, при необходимости подгружаем spice библиотеки компонентов и обученные модели с github¶

код CreateCVC базируется на библиотеке MySpice¶

пример использования CreateCVC¶

как построена база?¶

препроцессор данных для моделей -¶

делает преобразование фурье, прореживание и прочие манипуляции со входными сигналами¶

Мотивация к добавлению в признаковое пространство конкретных алгебраических величин (закон Ома):¶

Итого - параметры R, C, L линейно связаны с сигналами V, I, их производными (изменениями от точке к точке) и интегралами (суммами)¶

Мотивация к использованию логарифмов:¶

Итого - можно взять логарифмы от исходных сигналов, их производных и интегралов, и линейная модель их свяжет с логарифмами искомых параметров R, C, L.¶

базовый класс схем¶

Реализация конкретных схем из резисторов, емкостей, индуктивностей и диодов¶

rank 1, rank 2¶

Реализация конкретных схем из резисторов, емкостей, индуктивностей и диодов¶

rank 3¶

тестирование разных схем¶

Класс для предсказания схемы по ВАХ¶

тестирование класса предсказания схем¶

Синтетические тесты - искуственно сгенерированные данные¶

Тесты на реальных приборных Данных¶

https://github.com/vasily84/PySpice_CurveTracer ¶